package com.aoyama.kafka.base.demo03;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * ConsumerAwareListenerErrorHandler 消费异常处理器
 *
 * 通过异常处理器，我们可以处理consumer在消费时发生的异常。
 *
 * 新建一个 ConsumerAwareListenerErrorHandler 类型的异常处理方法，
 * 用@Bean注入，BeanName默认就是方法名，然后我们将这个异常处理器的BeanName放到@KafkaListener注解的errorHandler属性里面，
 * 当监听抛出异常的时候，则会自动调用异常处理器，
 */
@Component
public class ConsumerErrorHandler03 {

    // 新建一个异常处理器，用@Bean注入
    @Bean
    public ConsumerAwareListenerErrorHandler consumerAwareErrorHandler() {
        return (message, exception, consumer) -> {
            System.out.println("消费异常："+message.getPayload());
            return null;
        };
    }

    /*
    程序中不能出现监听两个相同的topic，否则会报错

    如下面同时开启两个topic6监听，运行的话会抛出异常
     */

    /**
     * 将这个异常处理器的BeanName放到@KafkaListener注解的errorHandler属性里面
     * 用@Bean注入，BeanName默认就是方法名
     * @param record
     * @throws Exception
     */
    @KafkaListener(topics = "topic6", errorHandler = "consumerAwareErrorHandler")
    public void onMessage4(ConsumerRecord<?, ?> record) throws Exception {
        throw new Exception("简单消费-模拟异常");
    }

    // 同上存在topic会报错
    // 批量消费也一样，异常处理器的message.getPayload()也可以拿到各条消息的信息
//    @KafkaListener(topics = "topic6")
//    public void onMessage5(List<ConsumerRecord<?, ?>> records) throws Exception {
//        throw new Exception("批量消费-模拟异常");
//    }


    /**
     * 注意：批量消费应配置文件中开启
     * 批量消费也一样，异常处理器的message.getPayload()也可以拿到各条消息的信息
     * @param records
     * @throws Exception
     */
    @KafkaListener(topics = "topic7", errorHandler = "consumerAwareErrorHandler")
    public void onMessage5(List<ConsumerRecord<?, ?>> records) throws Exception {
        throw new Exception("批量消费-模拟异常");
    }


}
